package com.central.scheduler.worker.runner;

import com.central.common.enums.TaskType;
import com.central.common.model.ResourceInfo;
import com.central.scheduler.common.Constants;
import com.central.scheduler.common.enums.CommandType;
import com.central.scheduler.common.enums.Event;
import com.central.scheduler.common.enums.ExecutionStatus;
import com.central.common.properties.Property;
import com.central.scheduler.common.utils.*;
import com.central.scheduler.remote.command.Command;
import com.central.scheduler.remote.command.TaskExecuteAckCommand;
import com.central.scheduler.remote.command.TaskExecuteResponseCommand;
import com.central.scheduler.service.alert.AlertClientService;
import com.central.scheduler.service.utils.ProcessUtils;
import com.central.scheduler.worker.cache.ResponceCache;
import com.central.scheduler.worker.plugin.TaskPluginManager;
import com.central.scheduler.worker.processor.TaskCallbackService;
import com.central.scheduler.service.queue.entity.TaskExecutionContext;
import com.central.scheduler.spi.exception.PluginNotFoundException;
import com.central.scheduler.spi.task.AbstractTask;
import com.central.scheduler.spi.task.TaskAlertInfo;
import com.central.scheduler.spi.task.TaskChannel;
import com.central.scheduler.spi.task.TaskExecutionContextCacheManager;
import com.central.scheduler.spi.task.request.TaskRequest;
import com.github.rholder.retry.RetryException;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.Delayed;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static java.util.Calendar.DAY_OF_MONTH;

/**
 * @author Tindy
 * @date 2021/10/29
 * @describe task scheduler thread
 */
public class TaskExecuteThread implements Runnable, Delayed {

    /**
     * logger
     */
    private final Logger logger = LoggerFactory.getLogger(TaskExecuteThread.class);
    /**
     * task instance
     */
    private TaskExecutionContext taskExecutionContext;
    /**
     * task callback service
     */
    private TaskCallbackService taskCallbackService;

    /**
     * taskExecutionContextCacheManager
     */
    private TaskExecutionContextCacheManager taskExecutionContextCacheManager;

    /**
     * abstract task
     */
    private AbstractTask task;

    /**
     * alert client server
     */
    private AlertClientService alertClientService;

    private TaskPluginManager taskPluginManager;

    public TaskExecuteThread(TaskExecutionContext taskExecutionContext,
                             TaskCallbackService taskCallbackService,
                             AlertClientService alertClientService,
                             TaskPluginManager taskPluginManager) {
        this.taskExecutionContext = taskExecutionContext;
        this.taskCallbackService = taskCallbackService;
        this.alertClientService = alertClientService;
        this.taskPluginManager = taskPluginManager;
    }

    @Override
    public void run() {

        TaskExecuteResponseCommand responseCommand = new TaskExecuteResponseCommand(taskExecutionContext.getTaskInstanceId(), taskExecutionContext.getProcessInstanceId());
        try {
            logger.info("script path : {}", taskExecutionContext.getExecutePath());

            if (taskExecutionContext.getStartTime() == null) {
                taskExecutionContext.setStartTime(new Date());
            }
            if (taskExecutionContext.getCurrentExecutionStatus() != ExecutionStatus.RUNNING_EXECUTION) {
                changeTaskExecutionStatusToRunning();
            }
            logger.info("the task begins to execute. task instance id: {}", taskExecutionContext.getTaskInstanceId());

            int dryRun = taskExecutionContext.getDryRun();
            // copy hdfs/minio file to local
            if (dryRun == Constants.DRY_RUN_FLAG_NO) {
                downloadResource(taskExecutionContext.getExecutePath(),
                        taskExecutionContext.getResources(),
                        logger);
            }

            taskExecutionContext.setEnvFile(CommonUtils.getSystemEnvPath());
            taskExecutionContext.setDefinedParams(getGlobalParamsMap());

            taskExecutionContext.setTaskAppId(String.format("%s_%s",
                    taskExecutionContext.getProcessInstanceId(),
                    taskExecutionContext.getTaskInstanceId()));

            preBuildBusinessParams();

            TaskChannel taskChannel = taskPluginManager.getTaskChannelMap().get(taskExecutionContext.getTaskType().name());
            if (null == taskChannel) {
                throw new PluginNotFoundException(String.format("%s Task Plugin Not Found,Please Check Config File.", taskExecutionContext.getTaskType()));
            }
            TaskRequest taskRequest = JSONUtils.parseObject(JSONUtils.toJsonString(taskExecutionContext), TaskRequest.class);
            String taskLogName = LoggerUtils.buildTaskId(LoggerUtils.TASK_LOGGER_INFO_PREFIX,
                    taskExecutionContext.getProcessDefineCode(),
                    taskExecutionContext.getProcessDefineVersion(),
                    taskExecutionContext.getProcessInstanceId(),
                    taskExecutionContext.getTaskInstanceId());
            taskRequest.setTaskLogName(taskLogName);

            this.task = taskChannel.createTask(taskRequest);
            // task init
            this.task.init();
            //init varPool
            this.task.getParameters().setVarPool(taskExecutionContext.getVarPool());

            if (dryRun == Constants.DRY_RUN_FLAG_NO) {
                // task handle
                //因为sqltask会改变线程名称，导致后续的日志输出过滤器无法生效，所有需要保存原有的线程名称，带sqltask执行完后恢复改线程名称
                String currentThreadName = Thread.currentThread().getName();
                this.task.handle();
                //恢复线程名称
                Thread.currentThread().setName(currentThreadName);
                // task result process
                if (this.task.getNeedAlert()) {
                    sendAlert(this.task.getTaskAlertInfo());
                }
                responseCommand.setStatus(this.task.getExitStatus().getCode());
            } else {
                responseCommand.setStatus(ExecutionStatus.SUCCESS.getCode());
                task.setExitStatusCode(Constants.EXIT_CODE_SUCCESS);
            }
            responseCommand.setEndTime(new Date());
            responseCommand.setProcessId(this.task.getProcessId());
            responseCommand.setAppIds(this.task.getAppIds());
            responseCommand.setVarPool(this.task.getParameters().getVarPool());
            logger.info("task instance id : {},task final status : {}", taskExecutionContext.getTaskInstanceId(), this.task.getExitStatus());
        } catch (Throwable e) {

            logger.error("task scheduler failure", e);
            kill();
            responseCommand.setStatus(ExecutionStatus.FAILURE.getCode());
            responseCommand.setEndTime(new Date());
            responseCommand.setProcessId(task.getProcessId());
            responseCommand.setAppIds(task.getAppIds());
        } finally {
            TaskExecutionContextCacheManager.removeByTaskInstanceId(taskExecutionContext.getTaskInstanceId());
            ResponceCache.get().cache(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command(), Event.RESULT);
            taskCallbackService.sendResult(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command());
            clearTaskExecPath();
        }
    }
    /**
     * when task finish, clear execute path.
     */
    private void clearTaskExecPath() {
        logger.info("develop mode is: {}", CommonUtils.isDevelopMode());
        if (!CommonUtils.isDevelopMode()) {
            // get exec dir
            String execLocalPath = taskExecutionContext.getExecutePath();

            if (StringUtils.isEmpty(execLocalPath)) {
                logger.warn("task: {} exec local path is empty.", taskExecutionContext.getTaskName());
                return;
            }

            if ("/".equals(execLocalPath)) {
                logger.warn("task: {} exec local path is '/', direct deletion is not allowed", taskExecutionContext.getTaskName());
                return;
            }

            try {
                org.apache.commons.io.FileUtils.deleteDirectory(new File(execLocalPath));
                logger.info("exec local path: {} cleared.", execLocalPath);
            } catch (IOException e) {
                logger.error("delete exec dir failed : {}", e.getMessage(), e);
            }
        }
    }

    /**
     * kill task
     */
    public void kill() {
        if (task != null) {
            try {
                task.cancelApplication(true);
                ProcessUtils.killYarnJob(taskExecutionContext);
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
    }
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(DateUtils.getRemainTime(taskExecutionContext.getFirstSubmitTime(),
                taskExecutionContext.getDelayTime() * 60L), TimeUnit.SECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
        if (o == null) {
            return 1;
        }
        return Long.compare(this.getDelay(TimeUnit.MILLISECONDS), o.getDelay(TimeUnit.MILLISECONDS));
    }

    /**
     * get current TaskExecutionContext
     *
     * @return TaskExecutionContext
     */
    public TaskExecutionContext getTaskExecutionContext() {
        return this.taskExecutionContext;
    }

    /**
     * send an ack to change the status of the task.
     */
    private void changeTaskExecutionStatusToRunning() {
        taskExecutionContext.setCurrentExecutionStatus(ExecutionStatus.RUNNING_EXECUTION);
        Command ackCommand = buildAckCommand().convert2Command();
        try {
            RetryerUtils.retryCall(() -> {
                taskCallbackService.sendAck(taskExecutionContext.getTaskInstanceId(), ackCommand);
                return Boolean.TRUE;
            });
        } catch (ExecutionException | RetryException e) {
            logger.error(e.getMessage(), e);
        }
    }

    /**
     * build ack command.
     *
     * @return TaskExecuteAckCommand
     */
    private TaskExecuteAckCommand buildAckCommand() {
        TaskExecuteAckCommand ackCommand = new TaskExecuteAckCommand();
        ackCommand.setTaskInstanceId(taskExecutionContext.getTaskInstanceId());
        ackCommand.setStatus(taskExecutionContext.getCurrentExecutionStatus().getCode());
        ackCommand.setStartTime(taskExecutionContext.getStartTime());
        ackCommand.setLogPath(taskExecutionContext.getLogPath());
        ackCommand.setHost(taskExecutionContext.getHost());
        if (TaskType.SQL.equals(taskExecutionContext.getTaskType()) || TaskType.PROCEDURE.equals(taskExecutionContext.getTaskType())) {
            ackCommand.setExecutePath(null);
        } else {
            ackCommand.setExecutePath(taskExecutionContext.getExecutePath());
        }
        return ackCommand;
    }

    /**
     * download resource file
     *
     * @param execLocalPath execLocalPath
     * @param res res
     * @param logger logger
     */
    private void downloadResource(String execLocalPath, List<ResourceInfo> res, Logger logger) {
        if (CollectionUtils.isEmpty(res)) {
            return;
        }

        for (ResourceInfo resource : res) {
            String name = resource.getName();
            String fileUrl = resource.getFileUrl();
            File resFile = new File(execLocalPath, name);
            if (!resFile.exists()) {
                try {
                    //使用wget命令 下载文件
                    wgetCommand(resource.getFileUrl(),execLocalPath,resource.getName());
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                    throw new RuntimeException(e.getMessage());
                }
            } else {
                logger.info("file : {} exists ", resFile.getName());
            }
        }
    }

    private void wgetCommand(String url,String execLocalPath,String resourceName) throws IOException, InterruptedException {
        List<String> command = new LinkedList<>();
        //init process builder
        ProcessBuilder processBuilder = new ProcessBuilder();
        // setting up a working directory
        processBuilder.directory(new File(execLocalPath));
        // merge error information to standard output stream
        processBuilder.redirectErrorStream(true);
        // setting commands
        command.add("wget");
        command.addAll(Collections.emptyList());
        command.add(url);
        command.addAll(Collections.emptyList());
        command.add("-P");
        command.addAll(Collections.emptyList());
        command.add(execLocalPath+"/"+resourceName);
        // setting up user to run commands
        processBuilder.command(command);
        processBuilder.start().waitFor();
    }
    /**
     * get global paras map
     *
     * @return map
     */
    private Map<String, String> getGlobalParamsMap() {
        Map<String, String> globalParamsMap = new HashMap<>(16);

        // global params string
        List<Property> globalParams= taskExecutionContext.getGlobalParams();
        globalParamsMap.putAll(globalParams.stream().collect(Collectors.toMap(Property::getProp, Property::getValue)));
        return globalParamsMap;
    }
    private void sendAlert(TaskAlertInfo taskAlertInfo) {
        alertClientService.sendAlert(taskAlertInfo.getAlertGroupId(), taskAlertInfo.getTitle(), taskAlertInfo.getContent());
    }
    private void preBuildBusinessParams() {
        Map<String, Property> paramsMap = new HashMap<>();
        // replace variable TIME with $[YYYYmmddd...] in shell file when history run job and batch complement job
        if (taskExecutionContext.getScheduleTime() != null) {
            Date date = taskExecutionContext.getScheduleTime();
            if (CommandType.COMPLEMENT_DATA.getCode() == taskExecutionContext.getCmdTypeIfComplement()) {
                date = DateUtils.add(taskExecutionContext.getScheduleTime(), DAY_OF_MONTH, 1);
            }
            String dateTime = DateUtils.format(date, Constants.PARAMETER_FORMAT_TIME);
            Property p = new Property();
            p.setValue(dateTime);
            p.setProp(Constants.PARAMETER_DATETIME);
            paramsMap.put(Constants.PARAMETER_DATETIME, p);
        }
        taskExecutionContext.setParamsMap(paramsMap);
    }

}
